package com.smsc.headend.connector.service.impl;

import cn.hutool.core.date.SystemClock;
import cn.hutool.core.io.file.FileWriter;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.smsc.headend.common.utils.ProtoBufferUtil;
import com.smsc.headend.common.utils.RedisKeys;
import com.smsc.headend.common.utils.RedisUtils;
import com.smsc.headend.connector.bo.RemoteDataCollector;
import com.smsc.headend.connector.mapper.FtpServerMapper;
import com.smsc.headend.connector.service.FeignAssetManagementService;
import com.smsc.headend.connector.service.FtpServerService;
import com.smsc.headend.module.asset.entity.FtpServer;
import com.smsc.headend.module.collection.dto.FtpServerListDto;
import com.smsc.headend.module.data.dto.FtpCollectionDTO;
import com.smsc.headend.module.data.proto.FtpCollectionDtoProto;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.math.RandomUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.*;

/**
 * @author liangli
 * @date 2020/12/1
 */
@Service
@Slf4j
public class FtpServerServiceImpl implements FtpServerService {
    @Autowired
    FeignAssetManagementService assetManagementService;

    List<RemoteDataCollector> templateCollectorHolder = new CopyOnWriteArrayList<>();

    @Autowired
    KafkaTemplate kafkaTemplate;

    @Autowired
    FtpServerMapper ftpServerMapper;

    @Value("${collection.ftp.localfilePath}")
    String collectionFileLocalPath;

    @Autowired
    RedisUtils redisUtils;

    private ExecutorService pool = Executors.newSingleThreadExecutor();

    @Override
    public void updateLastConTv(Long ftpId) {
        FtpServer ftpServer = new FtpServer();
        ftpServer.setFtpId(ftpId);
        ftpServer.setTvLastCon(SystemClock.now() / 1000);
        this.ftpServerMapper.updateById(ftpServer);
    }

    @Override
    public void saveLocalCollectionFile(String fileName, byte[] fileBytes) {
        String filePath = (collectionFileLocalPath.endsWith("/") ? collectionFileLocalPath : collectionFileLocalPath + "/") + fileName;
        log.info("save to local :{}", filePath);
        FileWriter writer = new FileWriter(filePath);
        writer.write(fileBytes, 0, fileBytes.length);
    }

    @Override
    public void refreshCollectionServer() {
        FtpServerListDto ftpServerListDto = assetManagementService.getAllIdleDcuCollectionFtpServer();
        if (ftpServerListDto == null || CollectionUtils.isEmpty(ftpServerListDto.getFtpIdPathMap())) {
            log.info("no collection server setting in UDIS");
            return;
        }
        Set<Long> ftpIds = ftpServerListDto.getFtpIdPathMap().keySet();
        if (CollectionUtils.isEmpty(ftpIds)) {
            log.info("no collection server setting in UDIS");
            return;
        }

        List<FtpServer> ftpServers = ftpServerMapper.selectList(Wrappers.<FtpServer>lambdaQuery().in(FtpServer::getFtpId, ftpIds));
        for (FtpServer ftpServer : ftpServers) {
            if (ftpServer.getTvLastCon() != null && ftpServer.getTvLastCon() > SystemClock.now() / 1000 - 5 * 60) {
                log.debug("FtpServerServiceImpl ftpId={},lastCon={},now={}", ftpServer.getFtpId(), ftpServer.getTvLastCon(), SystemClock.now() / 1000 - 5 * 60);
                continue;
            }

            Long key = getFtpConnectionLock(ftpServer.getFtpId());
            if (key == null) {
                continue;
            }
            try {
                RemoteDataCollector remoteDataCollector = buildRemoteDataCollector(ftpServer, ftpServerListDto.getFtpIdPathMap().get(ftpServer.getFtpId()), this);
                remoteDataCollector.connect();
                templateCollectorHolder.remove(remoteDataCollector);
                templateCollectorHolder.add(remoteDataCollector);
            } catch (Exception e) {
                log.error("init remote collector fail", e);
            } finally {
                ftpCollectionUnlock(ftpServer.getFtpId(), key);
            }
        }
    }

    private RemoteDataCollector buildRemoteDataCollector(FtpServer ftpServer, List<String> filePath, FtpServerServiceImpl ftpServerService) {
        RemoteDataCollector dataCollector = new RemoteDataCollector();
        dataCollector.setHost(ftpServer.getNetAddr().split(":")[0]);
        dataCollector.setPort(Integer.valueOf(ftpServer.getNetAddr().split(":")[1]));
        dataCollector.setUserName(ftpServer.getUserName());
        dataCollector.setPassword(ftpServer.getPassword());
        dataCollector.setServerType(ftpServer.getFtpType());
        dataCollector.setCollectedFilePath(filePath);
        dataCollector.setFtpServerService(ftpServerService);
        dataCollector.setFtpId(ftpServer.getFtpId());
        return dataCollector;
    }

    Long getFtpConnectionLock(Long ftpId) {
        String lockKey = RedisKeys.getFtpConnectionLockKey(ftpId);
        Long random = RandomUtils.nextLong();
        boolean result = redisUtils.tryLockWithId(lockKey, String.valueOf(random), 30);
        if (!result) {
            log.info("ftp connectAndSendTask lock fail {}", ftpId);
            return null;
        }
        log.info("lock ftp server {}", ftpId);
        return random;
    }

    void ftpCollectionUnlock(Long ftpId, Long lockKey) {
        String key = RedisKeys.getFtpConnectionLockKey(ftpId);
        Object lockValue = redisUtils.get(key);
        if (lockValue == null) {
            return;
        }
        if (Long.valueOf(String.valueOf(lockValue)) == lockKey.longValue()) {
            redisUtils.del(key);
        }
    }

    @Override
    public void collectRemoteFiles() {
        if (CollectionUtils.isEmpty(templateCollectorHolder)) {
            log.info("no collector in CON");
            return;
        }
        log.info("collect: start collect, size={}", templateCollectorHolder.size());
        Iterator iterator = templateCollectorHolder.iterator();
        while (iterator.hasNext()) {
            Future future = pool.submit(() -> {
                try {
                    RemoteDataCollector remoteDataCollector = (RemoteDataCollector) iterator.next();
                    remoteDataCollector.collectData();
                } catch (Exception e) {
                    log.error("collect data fail", e);
                }
            });
            try {
                //如果线程长时间阻塞，终止等下一次运行
                future.get(10 * 60, TimeUnit.SECONDS);
            } catch (InterruptedException | ExecutionException | TimeoutException e) {
                log.debug("future exception={}", e.getMessage());
                future.cancel(true);
            } catch (Exception e) {
                log.error(" ftp get file error:", e);
            }
        }

        log.debug("collectRemoteFiles finish");
    }

    @Override
    public void sendCollectionData(FtpCollectionDTO dto) {
        log.info(dto.toString());
        kafkaTemplate.send(TaskKafkaTopic.FTP_COLLECTION_DATA_TO_UTE, ProtoBufferUtil.toProtoBuffer(dto, FtpCollectionDtoProto.FtpCollectionDto.class).toByteArray());
    }
}
